package com.shujia.flink.window

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

import java.time.Duration

object Demo1TimeWindow {
  def main(args: Array[String]): Unit = {


    /**
     * 时间窗口
     * 1、SlidingEventTimeWindows： 滑动的事件时间窗口
     * 2、SlidingProcessingTimeWindows：滑动的处理时间窗口
     * 3、TumblingEventTimeWindows：滚动的事件时间窗口
     * 4、TumblingProcessingTimeWindows：滚动的处理时间窗口
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    /**
     * java,1678756528000
     * java,1678756529000
     * java,1678756530000
     * java,1678756531000
     * java,1678756532000
     * java,1678756535000
     * java,1678756537000
     * java,1678756540000
     * java,1678756550000
     */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val eventDS: DataStream[(String, Long)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val word: String = split(0)
      val ts: Long = split(1).toLong
      (word, ts)
    })


    //创建水位线和事件时间的生产策略
    val ws: WatermarkStrategy[(String, Long)] = WatermarkStrategy
      .forBoundedOutOfOrderness[(String, Long)](Duration.ofSeconds(5))
      .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
        override def extractTimestamp(element: (String, Long), recordTimestamp: Long): Long = element._2
      })

    //1、指定时间字段和水位线
    val assDS: DataStream[(String, Long)] = eventDS.assignTimestampsAndWatermarks(ws)

    /**
     * 每隔5秒统计单词的数量，统计最近15秒
     */
    assDS
      .map(kv => (kv._1, 1))
      .keyBy(_._1)
      .window(SlidingEventTimeWindows.of(Time.seconds(15), Time.seconds(5)))
      .sum(1)
      .print()


    env.execute()
  }

}
